package com.udf.hive.udtf;

import com.google.common.collect.Lists;
import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;

import java.util.List;
import java.util.Objects;

/**
 * 自定义 UDTF 表函数
 * select src.id, mytable.col1, mytable.col2
 * from src lateral view explode_map(properties) mytable as col1, col2;
 *
 * @author huadong.qi
 * @date 2020/11/11
 */
public class UDTFDemo1 extends GenericUDTF {

    private String[] obj = new String[2];

    /**
     * 初始化
     * 构建一个StructObjectInspector类型用于输出
     * 其中struct的字段构成输出的一行
     * 字段名称不重要，因为它们将被用户提供的列别名覆盖
     *
     * @param argOIs
     * @return
     * @throws UDFArgumentException
     */
    @Override
    public StructObjectInspector initialize(StructObjectInspector argOIs) throws UDFArgumentException {
        List<String> colNames = Lists.newLinkedList();
        colNames.add("key");
        colNames.add("value");
        List<ObjectInspector> resultTypes = Lists.newLinkedList();
        resultTypes.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        resultTypes.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        // 返回列名和列类型
        return ObjectInspectorFactory.getStandardStructObjectInspector(colNames, resultTypes);
    }

    /**
     * 真正的处理过程在process函数中，在process中，
     * 每一次forward()调用产生一行；如果产生多列可以将多个列的值放在一个数组中，
     * 然后将该数组传入到forward()函数
     *
     * @param args
     * @throws HiveException
     */
    @Override
    public void process(Object[] args) throws HiveException {
        if (Objects.isNull(args)) {
            return;
        }
        String args1 = args[0].toString();
        String[] ary1 = args1.split(",");
        String[] ary2 = args[1].toString().split(",");
        for (int i = 0; i < ary1.length; i++) {
            obj[0] = ary1[i];
            obj[1] = ary2[i];
            forward(obj);
        }

    }

    @Override
    public void close() throws HiveException {

    }
}
